package test;

import cn.hutool.core.date.DateUtil;
import com.huonan.utils.ExecutorUtil;
import lombok.SneakyThrows;
import org.junit.Test;
import test.entity.User;

import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * <a href="https://segmentfault.com/a/1190000014479792">...</a>
 */
public class AppExtractDataCompletableFutureEsExportApp {
    static ExecutorService executorService = ExecutorUtil.getExecutorService(3);

    @Test
    public void appThenAccept() {
        CompletableFuture.supplyAsync(() -> "productId-0001").thenAccept(product -> System.out.println("Got product detail from remote service " + product));

    }

    @Test
    public void appThenRun() {
        // thenRun() example
        CompletableFuture.supplyAsync(() -> {
            // Run some computation
            return "do something ,not return";
        }).thenRun(() -> {
            // Computation Finished.
            System.out.println("finished");
        });
    }

    @Test
    public void thenApply() {
        CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(2000);
                System.out.println(Thread.currentThread().getName());
            } catch (InterruptedException e) {
                throw new IllegalStateException(e);
            }
            return "Some Result";
        }).thenApply(result -> {
    /*
      Executed in the same thread where the supplyAsync() task is executed
      or in the main thread If the supplyAsync() task completes immediately (Remove sleep() call to verify)
    */
            System.out.println(Thread.currentThread().getName());
            return result + "->Processed Result";
        });

        System.out.println();
    }

    @SneakyThrows
    @Test
    public void thenApplyAsync() {
        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName());
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return "Some Result";
        }).thenApplyAsync(result -> {
            // Executed in a different thread from ForkJoinPool.commonPool()
            System.out.println(Thread.currentThread().getName());
            return result + "->Processed Result";
        });
        System.out.println(completableFuture.get());
    }

    @Test
    public void threadPoolThenApplyAsync() throws ExecutionException, InterruptedException {
        Executor executor = Executors.newFixedThreadPool(2);
        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName());
            return "Some result";
        }, executor).thenApplyAsync(result -> {
            // Executed in a thread obtained from the executor
            System.out.println(Thread.currentThread().getName());
            return result + "->Processed Result";
        }, executor);
        System.out.println(completableFuture.get());
    }


    @Test
    public void completableFuture() throws ExecutionException, InterruptedException {
        CompletableFuture<Double> result = getUsersDetail()
                .thenCompose(user -> getCreditRating());
        System.out.println(result.get());
    }


    @Test
    public void thenCombine() throws ExecutionException, InterruptedException {

        System.out.println("Retrieving weight.");
        CompletableFuture<Double> weightInKgFuture = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                throw new IllegalStateException(e);
            }
            return 65.0;
        });

        System.out.println("Retrieving height.");
        CompletableFuture<Double> heightInCmFuture = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                throw new IllegalStateException(e);
            }
            return 177.8;
        });

        System.out.println("Calculating BMI.");
        CompletableFuture<Double> combinedFuture = weightInKgFuture
                .thenCombine(heightInCmFuture, (weightInKg, heightInCm) -> {
                    Double heightInMeter = heightInCm / 100;
                    return weightInKg / (heightInMeter * heightInMeter);
                });

        System.out.println("Your BMI is - " + combinedFuture.get());

    }

    @Test
    public void thenCombineAllOf() throws ExecutionException, InterruptedException {

        List<String> webPageLinks = Arrays.asList("1", "2", "3");

        // Download contents of all the web pages asynchronously
        List<CompletableFuture<String>> pageContentFutures = webPageLinks.stream()
                .map(AppExtractDataCompletableFutureEsExportApp::downloadWebPage)
                .collect(Collectors.toList());


        // Create a combined Future using allOf()
        CompletableFuture<Void> allFutures = CompletableFuture.allOf(
                // pageContentFutures.toArray(new CompletableFuture[0])
                pageContentFutures.toArray(new CompletableFuture[0])

        );
        // When all the Futures are completed, call `future.join()` to get their results and collect the results in a list -
        CompletableFuture<List<String>> allPageContentsFuture = allFutures.thenApply(v -> pageContentFutures.stream()
                .map(CompletableFuture::join)
                .collect(Collectors.toList()));

        allPageContentsFuture.get().forEach(System.out::println);
    }


    @Test
    public void anyOf() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                throw new IllegalStateException(e);
            }
            return "Result of Future 1";
        });

        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                throw new IllegalStateException(e);
            }
            return "Result of Future 2";
        });

        CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                throw new IllegalStateException(e);
            }
            return "Result of Future 3";
        });

        CompletableFuture<Object> anyOfFuture = CompletableFuture.anyOf(future1, future2, future3);

        System.out.println(anyOfFuture.get()); // Result of Future 2
    }

    @Test
    public void tt() throws ExecutionException, InterruptedException {
        String str = "变量";
        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> str);
        System.out.println(completableFuture.get());
    }

    @Test
    public void handle() throws ExecutionException, InterruptedException {
        int age = -1;
        CompletableFuture<String> maturityFuture = CompletableFuture.supplyAsync(() -> {
            if (0 > age) {
                throw new IllegalArgumentException("Age can not be negative");
            }
            if (age > 18) {
                return "Adult";
            } else {
                return "Child";
            }
        }).handle((res, ex) -> {
            if (ex != null) {
                System.out.println("Oops! We have an exception - " + ex.getMessage());
                return "Unknown!";
            }
            return res;
        });

        System.out.println("Maturity : " + maturityFuture.get());
    }

    private static CompletableFuture<String> downloadWebPage(String pageLink) {
        return CompletableFuture.supplyAsync(() -> {
            // Code to download and return the web page's content
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return String.valueOf(DateUtil.millisecond(new Date()));
        }, executorService);
    }

    CompletableFuture<User> getUsersDetail() {
        return CompletableFuture.supplyAsync(() -> {
            User user = new User();
            user.setUserId("userId=001");
            return user;
        });
    }

    CompletableFuture<Double> getCreditRating() {
        return CompletableFuture.supplyAsync(() -> 0.1);
    }

    @SneakyThrows
    @Test
    public void appFuture() {

        System.out.println("Future start at:" + DateUtil.now());
        Future<?> future = executorService.submit(() -> {
            try {
                Thread.sleep(2000);
                System.out.println(Thread.currentThread().getName());
                System.out.println("Future finished at:" + DateUtil.now());
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
        //阻塞
        future.get();
        System.out.println("Main:" + Thread.currentThread().getName());
        System.out.println("主线程执行:" + DateUtil.now());
        Thread.sleep(2000);

        System.out.println("=============");
        System.out.println("CompletableFuture start at:" + DateUtil.now());
        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            System.out.println(Thread.currentThread().getName());
            System.out.println( "CompletableFuture 执行中 at:" + DateUtil.now());
            return "completableFuture finished ";
            //thenApplyAsync 异步执行，在不同线程：ForkJoinPool执行
            //thenApply ：和前一个执行同步执行，在相同线程
        },executorService).thenApply((r) -> {
            System.out.println(Thread.currentThread().getName());
            System.out.println(r + " at:" + DateUtil.now());
            f1(3);
            return r + " at: " + DateUtil.now();
        });
        //阻塞
        //completableFuture.get();
        //不会阻塞
        completableFuture.whenComplete((r, e) -> {
            System.out.println(Thread.currentThread().getName());
            System.out.println("异步执行:" + r);
        });
        System.out.println("Main:" + Thread.currentThread().getName());
        System.out.println("主线程执行:" + DateUtil.now());
        Thread.sleep(5000);
    }

    @SneakyThrows
    public void f1(int i)  {
        Thread.sleep((5 - i) * 1000L);
        System.out.println("输出:" + i);
    }
}
